package com.baosight.xdata.redis;

import io.lettuce.core.*;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.sync.RedisCommands;
import io.lettuce.core.models.stream.PendingMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Map;

public class TssLettuceForSingleNodeUtil {
    private static final Logger logger = LoggerFactory.getLogger(TssLettuceForSingleNodeUtil.class);


    private static class RedisConnectionFactory {
        private static RedisClient redisClient;
        private static StatefulRedisConnection<String, String> connection;

        public static void init(RedisURI uris) {
            logger.info("Initialize redis cluster connection [{}]", uris.toString());
            redisClient = RedisClient.create(uris);
            connection = redisClient.connect();
            logger.info("Finish to initialize redis cluster connection ");
        }

        public static StatefulRedisConnection<String, String> getConnection(){
            if(null == redisClient || null == connection ){
                throw new InternalError("redis cluster initialization not finish");
            }
            return connection;
        }

        public static RedisClient getClient(){
            if(null == redisClient || null == connection ){
                throw new InternalError("redis cluster initialization not finish");
            }
            return redisClient;
        }
    }

    public static void init(RedisURI uris) {
        TssLettuceForSingleNodeUtil.RedisConnectionFactory.init(uris);
    }

    public static void close(){
        if (TssLettuceForSingleNodeUtil.RedisConnectionFactory.getConnection() != null) {
            TssLettuceForSingleNodeUtil.RedisConnectionFactory.getConnection().close();
        }
        if (TssLettuceForSingleNodeUtil.RedisConnectionFactory.getClient() != null) {
            TssLettuceForSingleNodeUtil.RedisConnectionFactory.getClient().shutdown();
        }
        logger.info("redisClient shutdown in jvmHook.");
    }

    public static String xadd(String key, Map object) {
        RedisCommands<String, String> commands = TssLettuceForSingleNodeUtil.RedisConnectionFactory.getConnection().sync();
        return commands.xadd(key, object);
    }

    public static List<StreamMessage<String, String>> xread(XReadArgs args, XReadArgs.StreamOffset streamOffset) {
        RedisCommands<String, String> commands = TssLettuceForSingleNodeUtil.RedisConnectionFactory.getConnection().sync();
        return commands.xread(args, streamOffset);
    }

    public static List<StreamMessage<String, String>> xreadGroup(String key, String consumer, String group) {
        RedisCommands<String, String> commands = TssLettuceForSingleNodeUtil.RedisConnectionFactory.getConnection().sync();
        return commands.xreadgroup(Consumer.from(group, consumer),
                XReadArgs.StreamOffset.lastConsumed(key));
    }

    public static List<StreamMessage<String, String>> xreadGroupWithArgs(String key, String group, String consumer, XReadArgs args) {
        RedisCommands<String, String> commands = TssLettuceForSingleNodeUtil.RedisConnectionFactory.getConnection().sync();
        return commands.xreadgroup(Consumer.from(group, consumer), args,
                XReadArgs.StreamOffset.lastConsumed(key));
    }


    public static void xack(String key, StreamMessage<String, String> message, String group, String consumer) {
        RedisCommands<String, String> commands = TssLettuceForSingleNodeUtil.RedisConnectionFactory.getConnection().sync();
        commands.xack(key, group, message.getId());
        logger.info("finish to ack  message for stream [{}] group [{}] consumer [{}] messageId [{}]", key,group, consumer, message.getId() );
    }

    public static void xackWithBatch(String key, String group, String consumer, List<String> messageIdsList) {
        RedisCommands<String, String> commands = TssLettuceForSingleNodeUtil.RedisConnectionFactory.getConnection().sync();
        String[] messagesId = messageIdsList.toArray(new String[messageIdsList.size()]);
        commands.xack(key, group, messagesId);
        logger.info("finish to ack messages for stream [{}] group [{}] consumer [{}]  ack message size [{}]", key, group, consumer, messageIdsList.size());
    }

    public static void xdelWithBatch(String key, List<String> messageIdsList) {
        RedisCommands<String, String> commands = TssLettuceForSingleNodeUtil.RedisConnectionFactory.getConnection().sync();
        String[] messagesId = messageIdsList.toArray(new String[messageIdsList.size()]);
        commands.xdel(key, messagesId);
        logger.info(String.format("stream[%s] , del message size [%d]", key, messageIdsList.size()));
    }

    public static void xdel(String key, String group, String consumer, StreamMessage<String, String> message) {
        RedisCommands<String, String> commands = TssLettuceForSingleNodeUtil.RedisConnectionFactory.getConnection().sync();
        commands.xdel(key, message.getId());
        logger.info("finish to delete message for stream [{}] group [{}] consumer [{}] messageId [{}]", key, group, consumer, message.getId() );

    }

    public static String getInfo() {
        RedisCommands<String, String> commands = TssLettuceForSingleNodeUtil.RedisConnectionFactory.getConnection().sync();
        return commands.info();
    }

    public static String getInfo(String val) {
        RedisCommands<String, String> commands = TssLettuceForSingleNodeUtil.RedisConnectionFactory.getConnection().sync();
        return commands.info(val);
    }

    public static String getClusterInfo() {
        RedisCommands<String, String> commands = TssLettuceForSingleNodeUtil.RedisConnectionFactory.getConnection().sync();
        return commands.clusterInfo();
    }

    public static String getClusterNodes() {
        RedisCommands<String, String> commands = TssLettuceForSingleNodeUtil.RedisConnectionFactory.getConnection().sync();
        return commands.clusterNodes();
    }

    public static List<Object> getSlowlogGet() {
        RedisCommands<String, String> commands = TssLettuceForSingleNodeUtil.RedisConnectionFactory.getConnection().sync();
        return commands.slowlogGet();
    }

    public static List<Object> xinfoStream(String streamKey) {
        RedisCommands<String, String> commands = TssLettuceForSingleNodeUtil.RedisConnectionFactory.getConnection().sync();
        return commands.xinfoStream(streamKey);
    }

    public static List<Object> xinfoGroups(String streamKey) {
        RedisCommands<String, String> commands = TssLettuceForSingleNodeUtil.RedisConnectionFactory.getConnection().sync();
        return commands.xinfoGroups(streamKey);
    }

    public static List<String> keys(String streamKey) {
        RedisCommands<String, String> commands = TssLettuceForSingleNodeUtil.RedisConnectionFactory.getConnection().sync();
        return commands.keys(streamKey);
    }

    public static List<PendingMessage> xpendingGroup(String key, String group, String consumer, Range range, Limit limit) {
        RedisCommands<String, String> commands = TssLettuceForSingleNodeUtil.RedisConnectionFactory.getConnection().sync();
        return commands.xpending(key, Consumer.from(group, consumer), range, limit);
    }

    public static List<Object> getConsumerInfo(String streamKey, String group) {
        RedisCommands<String, String> commands = TssLettuceForSingleNodeUtil.RedisConnectionFactory.getConnection().sync();
        return commands.xinfoConsumers(streamKey, group);
    }

}
